package com.klun.tjy.tjydata.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;

public abstract class HDFSClient {


	abstract int insert(String source, String path);

	abstract int query();

	abstract int del();

	final public int handle(String source, String path) {
		FileSystem connection = null;
		try {
			connection = getConnection();
			InputStream in = new FileInputStream(source);
			OutputStream out = connection.create(new Path(path));
			IOUtils.copyBytes(in, out, 4096, true);

		} catch (IOException e) {
			e.printStackTrace();
		} catch (URISyntaxException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return 0;
	}

	private FileSystem fs;
	private Configuration conf = new Configuration();

	private FileSystem getConnection() throws IOException, URISyntaxException, InterruptedException {
		conf.set("fs.defaultFS", "hdfs://ns1"); //hdfs地址
		conf.set("dfs.nameservices", "ns1");
		conf.set("dfs.ha.namenodes.ns1", "nn1,nn2");
		conf.set("dfs.namenode.rpc-address.ns1.nn1", "init1:9000");
		conf.set("dfs.namenode.rpc-address.ns1.nn2", "init2:9000");
		conf.set("dfs.client.failover.proxy.provider.ns1", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
		fs = FileSystem.get(new URI("hdfs://ns1"), conf, "root");
		return fs;
	}

	private void close() {
		try {
			fs.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}


}
